package com.sardine.base.hbase;

import com.sardine.base.hbase.constants.ConstantConf;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;

import java.io.IOException;

/**
 * 上海京东到家元信信息技术有限公司<br>
 * Created with IntelliJ IDEA.
 * Date:2017/11/20
 * Time:15:09
 * hbase 配置中心
 *
 * @author daifeng<br>
 *         daifeng@imdada.cn
 * @since JDK 1.7 <br>
 * Copyright (c) 2017 imdada System Incorporated All Rights Reserved.
 */
public class HDJDataSource implements HDataSource, InitializingBean {
    private static final Logger logger = LoggerFactory.getLogger(HDJDataSource.class);

    /**
     * 每次从服务器scan获取的数据量
     */
    private String writeBuffer = ConstantConf.HBASE_CLIENT_WRITE_BUFFER;
    /**
     * 客户端写缓存的大小,这里配置了5M
     */
    private String scannerCaching = ConstantConf.HBASE_CLIENT_SCANNER_CACHING;
    /**
     * 是否切换到备份集群
     */
    private String slaveClusterSwitch = ConstantConf.SLAVE_CLUSTER_SWITCH;

    /**
     * HBase申请的erp
     */
    private String erp;

    /**
     * HBase的 实例编号
     */
    private String instanceName;
    /**
     * 表空间名称
     */
    private String tableSpace;

    /**
     * HBase的accesskey值
     */
    private String accesskey;
    /**
     * hbase configuration
     */
    private HConfiguration confg = null;

    /**
     * 初始化
     */
    private void init() {
        if (confg == null) {
            synchronized (this) {
                if (confg == null) {
                    confg = new HConfiguration();
                    confg.set("bdp.hbase.erp", getErp());//你的erp
                    confg.set("bdp.hbase.instance.name", getInstanceName());//申请的实例名称
                    confg.set("bdp.hbase.accesskey", getAccesskey());//实例对应的accesskey，请妥善保管你的AccessKey
                    confg.set("hbase.client.scanner.caching", getScannerCaching());
                    confg.set("hbase.client.write.buffer", getWriteBuffer());
                    confg.set("slave.cluster.switch", getSlaveClusterSwitch());
                    confg.setTableSpace(tableSpace);
                    logger.info("hbase 配置信息构建成功.");
                }
            }
        }
    }

    /**
     * 得到配置文件
     *
     * @return
     */
    @Override
    public HConfiguration getConfig() {
        return confg;
    }

    /**
     * 获取hbase的连接
     *
     * @return
     */
    @Override
    public Connection getConnection() {
        //第二种方式：代码中指定
        Connection connection = null;
        try {
            connection = ConnectionFactory.createConnection(confg);
            logger.info("连接hbase connection 创建成功...");
        } catch (IOException e) {
            logger.error("创建hbse连接失败，", e);
        }
        return connection;
    }
    @Override
    public void afterPropertiesSet() throws Exception {
        if (getErp() == null) {
            throw new RuntimeException("HBaseConfiguration erp 属性必填 ");
        }
        if (getInstanceName() == null) {
            throw new RuntimeException("HBaseConfiguration 实例编号属性必填 ");
        }

        if (getTableSpace() == null) {
            throw new RuntimeException("HBaseConfiguration 表空间名称属性必填 ");
        }

        if (getAccesskey() == null) {
            throw new RuntimeException("HBaseConfiguration accesskey 属性必填 ");
        }
        init();

    }

    public String getWriteBuffer() {
        return writeBuffer;
    }

    public void setWriteBuffer(String writeBuffer) {
        this.writeBuffer = writeBuffer;
    }

    public String getScannerCaching() {
        return scannerCaching;
    }

    public void setScannerCaching(String scannerCaching) {
        this.scannerCaching = scannerCaching;
    }

    public String getSlaveClusterSwitch() {
        return slaveClusterSwitch;
    }

    public void setSlaveClusterSwitch(String slaveClusterSwitch) {
        this.slaveClusterSwitch = slaveClusterSwitch;
    }

    public String getErp() {
        return erp;
    }

    public void setErp(String erp) {
        this.erp = erp;
    }

    public String getInstanceName() {
        return instanceName;
    }

    public void setInstanceName(String instanceName) {
        this.instanceName = instanceName;
    }

    public String getAccesskey() {
        return accesskey;
    }

    public void setAccesskey(String accesskey) {
        this.accesskey = accesskey;
    }


    public String getTableSpace() {
        return tableSpace;
    }

    public void setTableSpace(String tableSpace) {
        this.tableSpace = tableSpace;
    }


}
